﻿using System;
using System.Collections.Generic;
using System.Linq;
using Amazon.DynamoDBv2;
using Amazon.DynamoDBv2.DocumentModel;
using Amazon.DynamoDBv2.Model;
using AmazonDynamoDBStub.Extensions;
using AmazonDynamoDBStub.Extensions.Internal;
using AmazonDynamoDBStub.Helpers;
using AmazonDynamoDBStub.Models;
using AmazonDynamoDBStub.Repositories;

namespace AmazonDynamoDBStub
{
    public class AmazonDynamoDBClient : AmazonDynamoDB
    {
        private readonly IDynamoDBTableRepository dynamoDBTableRepository;
        private readonly IDynamoDBAttributeRepository dynamoDBAttributeRepository;

        public AmazonDynamoDBClient(
            IDynamoDBTableRepository dynamoDBTableRepository,
            IDynamoDBAttributeRepository dynamoDBAttributeRepository)
        {
            this.dynamoDBTableRepository = dynamoDBTableRepository;
            this.dynamoDBAttributeRepository = dynamoDBAttributeRepository;
        }

        public GetItemResponse GetItem(GetItemRequest getItemRequest)
        {
            var tableName = getItemRequest.TableName;
            var table = dynamoDBTableRepository.FindBy(tableName);

            if (table == null)
            {
                ExceptionHelper.ThrowResourceNotFoundException();
            }

            var hashKeyAttribute = getItemRequest.Key.GetKeyValuePair(table.HashKey.AttributeName);
            var rangeKeyAttribute = getItemRequest.Key.GetKeyValuePair(table.RangeKey.AttributeName);
            var predicate = ExpressionBuilder.BuildKeyAttributeExpression(hashKeyAttribute, rangeKeyAttribute);
            var existingItem = dynamoDBAttributeRepository.GetItem(tableName, predicate);

            var response = new GetItemResponse();
            response.GetItemResult.Item = existingItem != null ? existingItem.ToItemDictionary() : null;
            return response;
        }

        public QueryResponse Query(QueryRequest queryRequest)
        {
            var tableName = queryRequest.TableName;
            var table = dynamoDBTableRepository.FindBy(tableName);

            if (table == null)
            {
                ExceptionHelper.ThrowResourceNotFoundException();
            }

            if (table.RangeKey.AttributeName == null)
            {
                ExceptionHelper.ThrowAmazonDynamoDBException();
            }

            var hashKeyCondition = queryRequest.KeyConditions.GetKeyValuePair(table.HashKey.AttributeName);
            var rangeKeyCondition = queryRequest.KeyConditions.GetKeyValuePair(table.RangeKey.AttributeName);

            var predicate = ExpressionBuilder.BuildConditionExpression(hashKeyCondition, rangeKeyCondition);
            var existingItems = dynamoDBAttributeRepository.FindItems(tableName, predicate);

            var response = new QueryResponse();
            var items = existingItems.Select(item => item.ToItemDictionary()).ToList();
            response.QueryResult.Items = items;
            response.QueryResult.Count = items.Count();
            response.QueryResult.LastEvaluatedKey = null;
            return response;
        }

        public UpdateItemResponse UpdateItem(UpdateItemRequest updateItemRequest)
        {
            var tableName = updateItemRequest.TableName;
            var table = dynamoDBTableRepository.FindBy(tableName);

            if (table == null)
            {
                ExceptionHelper.ThrowResourceNotFoundException();
            }

            var hashKeyAttribute = updateItemRequest.Key.GetKeyValuePair(table.HashKey.AttributeName);
            var rangeKeyAttribute = updateItemRequest.Key.GetKeyValuePair(table.RangeKey.AttributeName);
            var predicate = ExpressionBuilder.BuildKeyAttributeExpression(hashKeyAttribute, rangeKeyAttribute);
            var existingItem = dynamoDBAttributeRepository.GetItem(tableName, predicate);

            if (updateItemRequest.AttributeUpdates.Select(attr => attr.Value.Value).Any(AttributeValueExtensions.IsNullOrEmpty))
            {
                ExceptionHelper.ThrowAmazonDynamoDBException();
            }

            foreach (var attribute in updateItemRequest.AttributeUpdates)
            {
                AttributeValue attributeValue = attribute.Value.Value;

                // Visit: http://docs.aws.amazon.com/amazondynamodb/latest/APIReference//API_UpdateItem.html#API_UpdateItem_RequestSyntax
                if (existingItem != null)
                {
                    #region If an item with the specified Key is found in the table:

                    DynamoDBAttribute existingAttribute = 
                        existingItem.SingleOrDefault(attr => attr.AttributeName == attribute.Key);

                    // Allowed Values: ADD, PUT, DELETE
                    switch (attribute.Value.Action)
                    {
                        case "PUT":
                            if (existingAttribute == null)
                            {
                                var attributeToAdd = DynamoDBAttributeFactory.Create(tableName, existingItem.Key, attribute);
                                this.dynamoDBAttributeRepository.Add(attributeToAdd);
                            }
                            else
                            {
                                existingAttribute.AttributeValue = attributeValue.GetAttributeValue();
                                existingAttribute.AttributeType = attributeValue.GetAttributeType();
                                this.dynamoDBAttributeRepository.Update(existingAttribute);
                            }

                            break;

                        case "DELETE":
                            if (attribute.Value == null)
                            {
                                // If no value is specified.

                                if (existingAttribute != null)
                                {
                                    this.dynamoDBAttributeRepository.Remove(existingAttribute);
                                }
                            }
                            else
                            {
                                if (!attributeValue.IsArray())
                                {
                                    if (existingAttribute != null && existingAttribute.AttributeType == attributeValue.GetAttributeType())
                                    {
                                        this.dynamoDBAttributeRepository.Remove(existingAttribute); // TOCHECK
                                    }
                                }

                                if (attributeValue.IsArray() && existingAttribute.IsArray())
                                {
                                    // If a set of values is specified, 
                                    // then those values are subtracted from the old set.

                                    if (existingAttribute.AttributeType != attributeValue.GetAttributeType())
                                    {
                                        ExceptionHelper.ThrowAmazonDynamoDBException();
                                    }

                                    var primitiveListToRemove = DocumentHelper.AttributeValueToDynamoDBEntry(attributeValue).AsPrimitiveList();

                                    if (!primitiveListToRemove.Entries.Any())
                                    {
                                        ExceptionHelper.ThrowAmazonDynamoDBException();
                                    }

                                    var primitiveListFromExistingAttribute = DocumentHelper.AttributeValueToDynamoDBEntry(
                                        new AttributeValue().WithValue(existingAttribute.AttributeValue, existingAttribute.AttributeType)).AsPrimitiveList();

                                    var primitivelistResult = new PrimitiveList(primitiveListFromExistingAttribute.Type);
                                    primitivelistResult.Entries = primitiveListFromExistingAttribute.Entries.Except(primitiveListToRemove.Entries).ToList();
                                    existingAttribute.AttributeValue = primitivelistResult.ConvertToAttributeValue().GetAttributeValue();
                                }
                            }

                            break;

                        case "ADD":
                            if (existingAttribute == null)
                            {
                                // If the attribute does not already exist, 
                                // then the attribute and its values are added to the item.

                                var attributeToAdd = DynamoDBAttributeFactory.Create(tableName, existingItem.Key, attribute);
                                this.dynamoDBAttributeRepository.Add(attributeToAdd);
                            }
                            else
                            {
                                if (existingAttribute.AttributeType == "N" && attributeValue.GetAttributeType() == "N")
                                {
                                    // If the existing attribute is a number, and if Value is also a number, 
                                    // then the Value is mathematically added to the existing attribute.

                                    var left = new Primitive(existingAttribute.AttributeValue, true);
                                    var right = DocumentHelper.AttributeValueToDynamoDBEntry(attributeValue).AsPrimitive();
                                    var result = left.Add(right);

                                    existingAttribute.AttributeValue = result.ConvertToAttributeValue().GetAttributeValue();

                                }
                                else if (existingAttribute.IsArray() && attributeValue.IsArray())
                                {
                                    // If the existing data type is a set, and if the Value is also a set, 
                                    // then the Value is added to the existing set.

                                    if (existingAttribute.AttributeType != attributeValue.GetAttributeType())
                                    {
                                        ExceptionHelper.ThrowAmazonDynamoDBException();
                                    }

                                    var primitiveListToAdd = DocumentHelper.AttributeValueToDynamoDBEntry(attributeValue).AsPrimitiveList();

                                    if (!primitiveListToAdd.Entries.Any())
                                    {
                                        ExceptionHelper.ThrowAmazonDynamoDBException();
                                    }

                                    var primitiveListFromExistingAttribute = DocumentHelper.AttributeValueToDynamoDBEntry(
                                        new AttributeValue().WithValue(existingAttribute.AttributeValue, existingAttribute.AttributeType)).AsPrimitiveList();
                                    var primitivelistResult = new PrimitiveList(primitiveListFromExistingAttribute.Type);
                                    primitivelistResult.Entries = primitiveListFromExistingAttribute.Entries.Union(primitiveListToAdd.Entries).ToList();
                                    existingAttribute.AttributeValue = primitivelistResult.ConvertToAttributeValue().GetAttributeValue();
                                }
                                else
                                    ExceptionHelper.ThrowAmazonDynamoDBException();
                            }

                            break;

                        default:
                            ExceptionHelper.ThrowAmazonDynamoDBException();
                            return null;
                    }

                    #endregion

                }
                else
                {
                    #region If no item with the specified Key is found:

                    switch (attribute.Value.Action)
                    {
                        case "PUT":
                            {
                                var newItem = CreateNewItemWithSpecifiedPrimaryKey(tableName, updateItemRequest.Key.ToArray());
                                existingItem = newItem;

                                var attributeToAdd = DynamoDBAttributeFactory.Create(tableName, existingItem.Key, attribute);
                                this.dynamoDBAttributeRepository.Add(attributeToAdd);

                                break;
                            }
                        case "DELETE":
                            break;
                        case "ADD":
                            {
                                string[] allowedAttributeTypes = new string[] { "N", "NS" };

                                if (!allowedAttributeTypes.Contains(attributeValue.GetAttributeType()))
                                {
                                    ExceptionHelper.ThrowAmazonDynamoDBException();
                                }

                                var newItem = CreateNewItemWithSpecifiedPrimaryKey(tableName, updateItemRequest.Key.ToArray());
                                existingItem = newItem;

                                var attributeToAdd = DynamoDBAttributeFactory.Create(tableName, existingItem.Key, attribute);
                                this.dynamoDBAttributeRepository.Add(attributeToAdd);

                                break;
                            }
                        default:
                            ExceptionHelper.ThrowAmazonDynamoDBException();
                            return null;
                    }

                    #endregion
                }
            }

            return new UpdateItemResponse();
        }

        public PutItemResponse PutItem(PutItemRequest putItemRequest)
        {
            // Visit: http://docs.aws.amazon.com/amazondynamodb/latest/APIReference//API_PutItem.html

            if (putItemRequest.TableName == null)
            {
                ExceptionHelper.ThrowAmazonDynamoDBException();
            }

            if (putItemRequest.Item.Count == 0)
            {
                ExceptionHelper.ThrowAmazonDynamoDBException();
            }

            var tableName = putItemRequest.TableName;
            var table = dynamoDBTableRepository.FindBy(tableName);

            if (table == null)
            {
                ExceptionHelper.ThrowResourceNotFoundException();
            }

            var hashKeyAtributeName = table.HashKey.AttributeName;
            var rangeKeyAttributeName = table.RangeKey.AttributeName;

            var hashKeyAttribute = putItemRequest.Item.GetKeyValuePair(hashKeyAtributeName);
            var rangeKeyAttribute = putItemRequest.Item.GetKeyValuePair(rangeKeyAttributeName);

            if (hashKeyAttribute.Equals(default(KeyValuePair<string, AttributeValue>)) ||
                (rangeKeyAttributeName != null && rangeKeyAttribute.Equals(default(KeyValuePair<string, AttributeValue>))))
            {
                ExceptionHelper.ThrowAmazonDynamoDBException();
            }

            foreach (var attribute in putItemRequest.Item)
            {
                #region Validation

                var entry = DocumentHelper.AttributeValueToDynamoDBEntry(attribute.Value);

                Primitive primitive = entry.AsPrimitive();
                PrimitiveList primitiveList = entry.AsPrimitiveList();

                if (primitive != null)
                {
                    if (primitive.Type == DynamoDBEntryType.String ||
                        primitive.Type == DynamoDBEntryType.Numeric)
                    {
                        if (String.IsNullOrEmpty(primitive.Value as string))
                        {
                            ExceptionHelper.ThrowAmazonDynamoDBException();
                        }
                    }
                    else
                    {
                        if ((primitive.Value as byte[]).Count() == 0)
                        {
                            ExceptionHelper.ThrowAmazonDynamoDBException();
                        }
                    }
                }

                if (primitiveList != null)
                {
                    if (primitiveList.Entries.Count == 0)
                    {
                        ExceptionHelper.ThrowAmazonDynamoDBException();
                    }
                }

                #endregion
            }

            var predicate = ExpressionBuilder.BuildKeyAttributeExpression(hashKeyAttribute, rangeKeyAttribute);
            var existingItem = dynamoDBAttributeRepository.GetItem(tableName, predicate);

            if (existingItem != null)
            {
                dynamoDBAttributeRepository.RemoveItem(existingItem);
            }

            var newItem = CreateNewItemWithSpecifiedPrimaryKey(tableName, hashKeyAttribute, rangeKeyAttribute);

            var attributesToAdd = putItemRequest.Item.ToDictionary(attr => attr.Key, attr => attr.Value);
            attributesToAdd.Remove(hashKeyAttribute.Key);
            if (rangeKeyAttribute.Key != null)
            {
                attributesToAdd.Remove(rangeKeyAttribute.Key);
            }

            foreach (var attribute in attributesToAdd)
            {
                var attributeToAdd = DynamoDBAttributeFactory.Create(tableName, newItem.Key, attribute);
                this.dynamoDBAttributeRepository.Add(attributeToAdd);
            }

            return new PutItemResponse();
        }

        public DescribeTableResponse DescribeTable(DescribeTableRequest describeTableRequest)
        {
            var tableName = describeTableRequest.TableName;
            var table = dynamoDBTableRepository.FindBy(tableName);

            if (table == null)
            {
                ExceptionHelper.ThrowResourceNotFoundException();
            }

            return new DescribeTableResponse
            {
                DescribeTableResult = new DescribeTableResult
                {
                    Table = new TableDescription
                    {
                        TableName = describeTableRequest.TableName,
                        KeySchema = new List<KeySchemaElement>
                        {
                            new KeySchemaElement
                            {
                                AttributeName = table.HashKey.AttributeName,
                                KeyType = "HASH"
                            },
                            new KeySchemaElement
                            {
                                AttributeName = table.RangeKey.AttributeName,
                                KeyType = "RANGE"
                            }
                        },
                        AttributeDefinitions = new List<AttributeDefinition>
                        {
                            new AttributeDefinition
                            {
                                AttributeName = table.HashKey.AttributeName,
                                AttributeType = table.HashKey.AttributeType,
                            },
                            new AttributeDefinition
                            {
                                AttributeName = table.RangeKey.AttributeName,
                                AttributeType = table.RangeKey.AttributeType
                            }
                        },
                        TableStatus = table.TableStatus,
                        ItemCount = dynamoDBAttributeRepository.GetItemCount(tableName),
                        CreationDateTime = table.CreationDateTime
                    }
                }
            };
        }

        public CreateTableResponse CreateTable(CreateTableRequest createTableRequest)
        {
            if (createTableRequest.TableName == null)
            {
                ExceptionHelper.ThrowAmazonDynamoDBException();
            }

            if (createTableRequest.AttributeDefinitions.Count == 0)
            {
                ExceptionHelper.ThrowAmazonDynamoDBException();
            }

            if (createTableRequest.ProvisionedThroughput == null)
            {
                ExceptionHelper.ThrowAmazonDynamoDBException();
            }

            if (createTableRequest.KeySchema.Count == 0)
            {
                ExceptionHelper.ThrowAmazonDynamoDBException();
            }

            string tableName = createTableRequest.TableName;

            var hashKeySchema = createTableRequest.KeySchema.Single(key => key.KeyType == "HASH");
            var rangeKeySchema = createTableRequest.KeySchema.SingleOrDefault(key => key.KeyType == "RANGE");

            string hashKeyAttributeName = hashKeySchema.AttributeName;
            string rangeKeyAttributeName = rangeKeySchema != null ? rangeKeySchema.AttributeName : null;

            var hashKeyAttributeDefinition = createTableRequest.AttributeDefinitions.Single(attr => attr.AttributeName == hashKeyAttributeName);
            var rangeKeyAttributeDefinition = rangeKeyAttributeName != null ? createTableRequest.AttributeDefinitions.Single(attr => attr.AttributeName == hashKeyAttributeName) : null;

            string hashKeyAttributeType = hashKeyAttributeDefinition.AttributeType;
            string rangeKeyAttributeType = rangeKeyAttributeDefinition != null ? rangeKeyAttributeDefinition.AttributeType : null;

            var dynamoDBKey = DynamoDBTableFactory.Create(tableName, hashKeyAttributeName, hashKeyAttributeType, rangeKeyAttributeName, rangeKeyAttributeType);

            dynamoDBTableRepository.Add(dynamoDBKey);

            return new CreateTableResponse();
        }

        public DeleteItemResponse DeleteItem(DeleteItemRequest deleteItemRequest)
        {
            if (deleteItemRequest.TableName == null)
            {
                ExceptionHelper.ThrowAmazonDynamoDBException();
            }

            var tableName = deleteItemRequest.TableName;
            var table = dynamoDBTableRepository.FindBy(tableName);

            if (table == null)
            {
                ExceptionHelper.ThrowResourceNotFoundException();
            }

            var hashKeyAtributeName = table.HashKey.AttributeName;
            var rangeKeyAttributeName = table.RangeKey.AttributeName;

            var hashKeyAttribute = deleteItemRequest.Key.GetKeyValuePair(table.HashKey.AttributeName);
            var rangeKeyAttribute = deleteItemRequest.Key.GetKeyValuePair(table.RangeKey.AttributeName);

            var predicate = ExpressionBuilder.BuildKeyAttributeExpression(hashKeyAttribute, rangeKeyAttribute);

            var existingItem = dynamoDBAttributeRepository.GetItem(tableName, predicate);
            if (existingItem != null)
            {
                dynamoDBAttributeRepository.RemoveItem(existingItem);
            }

            return new DeleteItemResponse();
        }

        public DeleteTableResponse DeleteTable(DeleteTableRequest deleteTableRequest)
        {
            if (deleteTableRequest.TableName == null)
            {
                ExceptionHelper.ThrowAmazonDynamoDBException();
            }

            var tableName = deleteTableRequest.TableName;
            var table = dynamoDBTableRepository.FindBy(tableName);

            if (table == null)
            {
                ExceptionHelper.ThrowResourceNotFoundException();
            }

            dynamoDBTableRepository.Remove(table);

            return new DeleteTableResponse();
        }

        public ListTablesResponse ListTables()
        {
            var tables = dynamoDBTableRepository.FindAll();
            var response = new ListTablesResponse();
            response.ListTablesResult.TableNames = tables.Select(t => t.TableName).ToList();
            return response;
        }

        public ListTablesResponse ListTables(ListTablesRequest listTablesRequest)
        {
            throw new NotImplementedException();
        }

        public ScanResponse Scan(ScanRequest scanRequest)
        {
            var tableName = scanRequest.TableName;
            var table = dynamoDBTableRepository.FindBy(tableName);

            if (table == null)
            {
                ExceptionHelper.ThrowResourceNotFoundException();
            }

            var predicate = ExpressionBuilder.BuildConditionExpression(scanRequest.ScanFilter.ToArray());
            var existingItems = dynamoDBAttributeRepository.FindItems(tableName, predicate);

            var response = new ScanResponse();
            var items = existingItems.Select(item => item.ToItemDictionary()).ToList();
            response.ScanResult.Items = items;
            response.ScanResult.Count = items.Count();
            response.ScanResult.LastEvaluatedKey = null;
            return response;
        }

        public UpdateTableResponse UpdateTable(UpdateTableRequest updateTableRequest)
        {
            throw new NotImplementedException();
        }

        public BatchGetItemResponse BatchGetItem(BatchGetItemRequest batchGetItemRequest)
        {
            throw new NotImplementedException();
        }

        public BatchWriteItemResponse BatchWriteItem(BatchWriteItemRequest batchWriteItemRequest)
        {
            throw new NotImplementedException();
        }

        #region Asynchronous Calls

        public IAsyncResult BeginBatchGetItem(BatchGetItemRequest batchGetItemRequest, AsyncCallback callback, object state)
        {
            throw new NotImplementedException();
        }

        public IAsyncResult BeginBatchWriteItem(BatchWriteItemRequest batchWriteItemRequest, AsyncCallback callback, object state)
        {
            throw new NotImplementedException();
        }

        public IAsyncResult BeginCreateTable(CreateTableRequest createTableRequest, AsyncCallback callback, object state)
        {
            throw new NotImplementedException();
        }

        public IAsyncResult BeginDeleteItem(DeleteItemRequest deleteItemRequest, AsyncCallback callback, object state)
        {
            throw new NotImplementedException();
        }

        public IAsyncResult BeginDeleteTable(DeleteTableRequest deleteTableRequest, AsyncCallback callback, object state)
        {
            throw new NotImplementedException();
        }

        public IAsyncResult BeginDescribeTable(DescribeTableRequest describeTableRequest, AsyncCallback callback, object state)
        {
            throw new NotImplementedException();
        }

        public IAsyncResult BeginGetItem(GetItemRequest getItemRequest, AsyncCallback callback, object state)
        {
            throw new NotImplementedException();
        }

        public IAsyncResult BeginListTables(ListTablesRequest listTablesRequest, AsyncCallback callback, object state)
        {
            throw new NotImplementedException();
        }

        public IAsyncResult BeginPutItem(PutItemRequest putItemRequest, AsyncCallback callback, object state)
        {
            throw new NotImplementedException();
        }

        public IAsyncResult BeginQuery(QueryRequest queryRequest, AsyncCallback callback, object state)
        {
            throw new NotImplementedException();
        }

        public IAsyncResult BeginScan(ScanRequest scanRequest, AsyncCallback callback, object state)
        {
            throw new NotImplementedException();
        }

        public IAsyncResult BeginUpdateItem(UpdateItemRequest updateItemRequest, AsyncCallback callback, object state)
        {
            throw new NotImplementedException();
        }

        public IAsyncResult BeginUpdateTable(UpdateTableRequest updateTableRequest, AsyncCallback callback, object state)
        {
            throw new NotImplementedException();
        }

        public BatchGetItemResponse EndBatchGetItem(IAsyncResult asyncResult)
        {
            throw new NotImplementedException();
        }

        public BatchWriteItemResponse EndBatchWriteItem(IAsyncResult asyncResult)
        {
            throw new NotImplementedException();
        }

        public CreateTableResponse EndCreateTable(IAsyncResult asyncResult)
        {
            throw new NotImplementedException();
        }

        public DeleteItemResponse EndDeleteItem(IAsyncResult asyncResult)
        {
            throw new NotImplementedException();
        }

        public DeleteTableResponse EndDeleteTable(IAsyncResult asyncResult)
        {
            throw new NotImplementedException();
        }

        public DescribeTableResponse EndDescribeTable(IAsyncResult asyncResult)
        {
            throw new NotImplementedException();
        }

        public GetItemResponse EndGetItem(IAsyncResult asyncResult)
        {
            throw new NotImplementedException();
        }

        public ListTablesResponse EndListTables(IAsyncResult asyncResult)
        {
            throw new NotImplementedException();
        }

        public PutItemResponse EndPutItem(IAsyncResult asyncResult)
        {
            throw new NotImplementedException();
        }

        public QueryResponse EndQuery(IAsyncResult asyncResult)
        {
            throw new NotImplementedException();
        }

        public ScanResponse EndScan(IAsyncResult asyncResult)
        {
            throw new NotImplementedException();
        }

        public UpdateItemResponse EndUpdateItem(IAsyncResult asyncResult)
        {
            throw new NotImplementedException();
        }

        public UpdateTableResponse EndUpdateTable(IAsyncResult asyncResult)
        {
            throw new NotImplementedException();
        }

        #endregion

        #region Dispose Pattern Implementation

        protected virtual void Dispose(bool disposing)
        {
            if (disposing)
            {
                dynamoDBTableRepository.Dispose();
                dynamoDBAttributeRepository.Dispose();
            }
        }

        public void Dispose()
        {
            this.Dispose(true);
            GC.SuppressFinalize(this);
        }

        ~AmazonDynamoDBClient()
        {
            this.Dispose(false);
        }

        #endregion

        #region Private Methods

        private IGrouping<Guid, DynamoDBAttribute> CreateNewItemWithSpecifiedPrimaryKey(string tableName, params KeyValuePair<string, AttributeValue>[] keyAttributes)
        {
            #region Validate keys

            keyAttributes = keyAttributes.GetValid().ToArray();
            var dynamoDBKey = dynamoDBTableRepository.FindBy(tableName);

            foreach (var keyAttribute in keyAttributes)
            {
                string attributeName = keyAttribute.Key;
                string attributeType = keyAttribute.Value != null ? keyAttribute.Value.GetAttributeType() : null;

                if ((dynamoDBKey.HashKey.AttributeName != attributeName || dynamoDBKey.HashKey.AttributeType != attributeType) &&
                    (dynamoDBKey.RangeKey.AttributeName != attributeName || dynamoDBKey.RangeKey.AttributeType != attributeType))
                {
                    ExceptionHelper.ThrowAmazonDynamoDBException();
                }
            }

            #endregion

            var item = dynamoDBAttributeRepository.CreateItem(tableName, keyAttributes);
            return item;
        }

        #endregion
    }
}
